package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo2Partition {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Demo2Partition")

    val sc = new SparkContext(conf)

    /**
     * 读取目录，如果读取的是目录，目录下不能有子目录
     *
     */


    /**
     * RDDd的分区数，分区数越多，并行度越高，在资源充足的情况下效率越高
     * 1、读取文件，默认等于切片的数量
     * 2、读取文件可以设置最新分区数minPartitions，
     * 控制rdd的分区数，，只能在切片数的基础上，增加分区数，不能减少分区数
     * 原则是需要保证每一个分区的数据量差不多
     *
     * 3、窄依赖的算子不能改变分区数，默认等于前一个rdd的分区数
     * 4、宽依赖算子分区数默认等于前一个RDD,也可以手动设置分区数
     *
     *
     */

    val linesRDD: RDD[String] = sc.textFile("data/words", 7)

    println(s"linesRDD分区数：${linesRDD.getNumPartitions}")

    val wordsRDD: RDD[String] = linesRDD.flatMap(_.split(","))

    println(s"wordsRDD分区数：${wordsRDD.getNumPartitions}")

    val kvRDD: RDD[(String, Iterable[String])] = wordsRDD.groupBy(w => w, numPartitions = 10)

    println(s"kvRDD分区数：${kvRDD.getNumPartitions}")

    val wordcountRDD: RDD[(String, Int)] = kvRDD.map(kv => (kv._1, kv._2.size))

    println(s"wordcountRDD分区数：${wordcountRDD.getNumPartitions}")

    wordcountRDD.saveAsTextFile("data/wc")


  }

}
